package com.jd.rapid.rpc.server;

import com.jd.rapid.rpc.codec.RpcRequest;
import com.jd.rapid.rpc.codec.RpcResponse;
import io.netty.channel.*;
import io.netty.util.concurrent.EventExecutorGroup;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cglib.reflect.FastClass;
import org.springframework.cglib.reflect.FastMethod;

import java.lang.reflect.InvocationTargetException;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
public class RpcServerHandler extends SimpleChannelInboundHandler<RpcRequest> {

    private Map<String,Object> handlerMap;

    /*****用于异步的提交连接请求的线程池************/
    private ThreadPoolExecutor executor=new ThreadPoolExecutor(16,16,
            60, TimeUnit.SECONDS,new ArrayBlockingQueue<>(65536));


    public RpcServerHandler(Map<String,Object> handlerMap){
        this.handlerMap=handlerMap;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcRequest request) throws Exception {
        //接收请求
        executor.submit(new Runnable() {
            @Override
            public void run() {
                RpcResponse response=new RpcResponse();
                response.setRequestId(request.getRequestId());
                try{
                    Object result= hanlde(request);
                    response.setResult(result);
                }catch (Throwable t){
                    response.setThrowable(t);
                    log.error("rpc service handle request Throwable:"+t);
                }
                //1、解析rpcRequest
                //2、从handerMap找到具体的接口(key)所绑定的实例
                //3、通过反射cglib调用具体方法，传递相关执行参数执行逻辑即可
                //4 返回响应信息给调用方

                //后置处理
                ctx.writeAndFlush(response)
                   .addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture channelFuture) throws Exception {
                                if(channelFuture.isSuccess()){
                                    // afterRpcHook
                                }
                    }
                });
            }
        });
    }

    /**
     *  解析request请求并且通过反射获取具体的本地服务实例后执行具体的方法
     * @param request
     * @return
     * @throws InvocationTargetException
     */
    private Object hanlde(RpcRequest request) throws InvocationTargetException {
         String className=request.getClassName();
//        Object serviceBean= handlerMap.get(className);
         Object serviceRef= handlerMap.get(className);
         Class<?> serviceClass=serviceRef.getClass();
         String methodName=request.getMethodName();
         Class<?>[] parameterTypes=request.getParamterType();
         Object[] params=request.getParamters();

         //JDK reflect

        //Cglib
        FastClass serviceFastClass=FastClass.create(serviceClass);
        FastMethod serviceFastMethod=serviceFastClass.getMethod(methodName,parameterTypes);
        return serviceFastMethod.invoke(serviceRef,params);
    }


    /**
     * 异常处理关闭连接
     * @param ctx
     * @param cause
     */
    public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause){
            log.error("server caught Throwable:"+cause);
            ctx.close();
    }
}
